package com.j.lemon.learn.flink.datastream;

import com.j.lemon.learn.flink.WordCount;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * 返回数据集前n个元素
 * @author lijunjun
 */
public class SplitOperator {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<WordCount> dataStream = env.fromElements(
                new WordCount("zhangsan", 1),
                new WordCount("zhangsan", 2),
                new WordCount("lisi", 2),
                new WordCount("wangwu", 2),
                new WordCount("lisi", 3));
        OutputTag<WordCount> zhangsan = new OutputTag<WordCount>("1"){private static final long serialVersionUID = 1L;};
        OutputTag<WordCount> other = new OutputTag<WordCount>("2"){private static final long serialVersionUID = 1L;};
        SingleOutputStreamOperator<WordCount> process = dataStream.process(new ProcessFunction<WordCount, WordCount>() {
            @Override
            public void processElement(WordCount wordCount, Context context, Collector<WordCount> collector) throws Exception {
                if ("zhangsan".equals(wordCount.getWord())) {
                    context.output(zhangsan, wordCount);
                } else {
                    context.output(other, wordCount);
                }
            }
        });
        DataStream<WordCount> zhangsanDs = process.getSideOutput(zhangsan);
        DataStream<WordCount> otherDs = process.getSideOutput(other);

        ConnectedStreams<WordCount, WordCount> connect = zhangsanDs.connect(otherDs);
        connect.map(new CoMapFunction<WordCount, WordCount, String>() {
            @Override
            public String map1(WordCount wordCount) throws Exception {
                return null;
            }

            @Override
            public String map2(WordCount wordCount) throws Exception {
                return null;
            }
        });
        zhangsanDs.printToErr();
        System.out.println("~~~");
        otherDs.print();
        System.out.println(env.getExecutionPlan());
        env.execute();


    }
}
